package com.ibm.cps.processors.window;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.log4j.Logger;
import org.joda.time.DateTime;

import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.ibm.cps.dft.FixedSizeQueue;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.cps.message.ParsedDataMessageBuilder;
import com.ibm.cps.processors.BasicProcessor;
import com.ibm.cps.processors.Info;
import com.ibm.cps.sql.expression.function.meta.ValueParam;
import com.ibm.datamodel.timeseries.TimeSeriesSchema;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.util.DataTypes;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.RandomUtil;
import com.ibm.util.exception.CPSException;

@Info(Name=MessageFields.COUNT_WINDOW_TYPE,Type=Info.ProcessorType.Other)
public class CountWindowProcessor extends BasicProcessor{
	private static final Logger logger = Logger.getLogger(CountWindowProcessor.class);

	private ITimeSeriesSchema parentSchema;
	private ITimeSeriesSchema outputSchema;
	@ValueParam
	private int count;
	@ValueParam
	private int step;
	private final Map<Comparable, FixedSizeQueue<ParsedDataMessage>> buffer = new HashMap<>();
	
	public CountWindowProcessor(String tenantId, String processorId, String source, ITimeSeriesSchema parentSchema) throws CPSException {
		super(tenantId, processorId, source, parentSchema);
		this.parentSchema = parentSchema;
	}
	
	@Override
	public void open() throws CPSException {
		this.outputSchema = createOutputSchema();
	}

	private ITimeSeriesSchema createOutputSchema() throws CPSException {
		// create an array for each field with size dimension
		ArrayNode arrayNode = JSONUtility.newArrayNode();
		byte[] types = parentSchema.getDataTypes();
		String[] names = parentSchema.getTagNames();
		for (int i = 2; i < types.length; i++) {
			String fieldName = names[i]+"_array";
			String fieldType = DataTypes.getDataType(types[i]);
			
// modify by gxj to support array
			ObjectNode node = JSONUtility.newObjectNode();
			node.put(MessageFields.NAME, fieldName);
			if (fieldType.equals("int")) {
				node.put(MessageFields.DATATYPE, "int[]");
			}else if (fieldType.equals("double")) {
				node.put(MessageFields.DATATYPE, "double[]");
			}else if (fieldType.equals("string")) {
				node.put(MessageFields.DATATYPE, "string[]");
			}else {
				throw new CPSException(ErrorCode.NO_IMPLEMENTATION_CODE, "data type not support for array :" + fieldType);
			}
			arrayNode.add(node);
		}
		String schemaId = RandomUtil.getRandomId();
		return new TimeSeriesSchema(tenantId, schemaId, arrayNode, (new DateTime()).getMillis());
	}

	//What if tsKey is null?
	@Override
	public Collection<ParsedDataMessage> execute(ParsedDataMessage message) throws CPSException {
		Comparable timeseriesKey = message.getTsKey();
		FixedSizeQueue<ParsedDataMessage> bufferForTimeseries = buffer.get(timeseriesKey);
		if (bufferForTimeseries == null) {
			bufferForTimeseries = new FixedSizeQueue<>(count);
			buffer.put(timeseriesKey, bufferForTimeseries);
		}
		if (!containsNullValue(message)) {
			bufferForTimeseries.add(message);
		}
		logger.info("==============Count Window input============");
		logger.info(Arrays.toString(message.getObjectValues()));
		if (bufferForTimeseries.isFull()) {
			ParsedDataMessage[] output = new ParsedDataMessage[] {};
			output = bufferForTimeseries.toArray(output);
			Comparable tsKey = output[0].getTsKey();
			long timestamp = output[0].getTimestamp();
			
			ParsedDataMessageBuilder builder = ParsedDataMessageBuilder.createBasicBuilder(outputSchema, tsKey, timestamp);
			for (int i = 2; i < parentSchema.getTagCount(); i++) {
				String originalFieldName = parentSchema.getTagNames()[i];
				Object[] tempFieldValue = new Object[output.length];
				for (int j = 0; j < output.length; j++) {
					Object value = output[j].getValue(i);
					tempFieldValue[j] = value;
				}
				builder = builder.append(originalFieldName, tempFieldValue);
			}
			for (int i = 0; i < step; i++) {
				bufferForTimeseries.remove();
			}
			List<ParsedDataMessage> result = new ArrayList<>();
			ParsedDataMessage parsedData = builder.toParsedDataMessage();
			result.add(parsedData);

			logger.info("========== Count Window output ===========");
			logger.info(Arrays.toString(outputSchema.getTagNames()));
			logger.info(Arrays.toString(parsedData.getObjectValues()));
			return result;
		} else {
			return null;
		}
	}

	@SuppressWarnings("rawtypes")
	private boolean containsNullValue(ParsedDataMessage message) {
		Object[] values = message.getObjectValues();
		for (Object v : values) {
			if (v == null) {
				logger.info("WindowProcessor: received null value in " + values);
				return true;
			}
		}
		return false;
	}

	@Override
	public ITimeSeriesSchema getTimeSeriesSchema() throws CPSException {
		return outputSchema;
	}

	@Override
	public void dispose() {
	}

}
